package com.miya.demo.support.kafka.consumer;

import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * kafka消费者handler
 *
 * @author CaiXiaowei
 * @date 2023/6/2
 */
@NoArgsConstructor
@AllArgsConstructor
@Component
@Slf4j
public class KafkaConsumerHandler {

    public static final String TOPIC = "topic.demo";

    @NonNull
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    /**
     * 单条消费
     * @param consumerRecord
     * @param ack
     */
    @KafkaListener(topics = {TOPIC}, errorHandler = "kafkaConsumerExceptionHandler")
    public void listen(ConsumerRecord<Object, Object> consumerRecord, Acknowledgment ack) {
        try {
            //用于测试异常处理
            log.info("KafkaConsumerHandler message:{}", consumerRecord.value());
            //手动确认
            ack.acknowledge();
        } catch (Exception e) {
            System.out.println("消费失败：" + e);

        }
    }

    /**
     * 批量消费: 自定义containerFactory, 且设置batchListener=true
     * @param consumerRecords
     * @param ack
     */
    @KafkaListener(topics = {TOPIC}, containerFactory = "kafkaListenerContainerFactory", errorHandler = "kafkaConsumerExceptionHandler")
    public void listenBatch(List<ConsumerRecord<Object, Object>> consumerRecords, Acknowledgment ack) {
        try {
            //用于测试异常处理
            log.info("KafkaConsumerHandler size: {}, message:{}", consumerRecords.size(), consumerRecords);
            //手动确认
            ack.acknowledge();
        } catch (Exception e) {
            System.out.println("消费失败：" + e);

        }
    }
}
